package org.raymond.iworks.study.mq.kafka.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.*;
import org.springframework.kafka.support.converter.JsonMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
@Configuration
@ConditionalOnProperty(name="kafka.sample", havingValue="demo4")
public class Demo4Config {
    private static final String KAFKA_SERVER_ADDR = "192.168.0.125:9094,192.168.0.125:9194,192.168.0.125:9294";

    @Bean
    public KafkaAdmin admin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER_ADDR);
        return new KafkaAdmin(configs);
    }
    @Bean
    public NewTopic topic() {
        log.info("topic demo4");
        return TopicBuilder.name("demo4")
                .partitions(1)
                .replicas(1)
                .build();
    }

    public ProducerFactory producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        return Demo1Config.producerConfigs(JsonSerializer.class);
    }

    @Bean
    public RecordMessageConverter converter() {
        return new JsonMessageConverter();
    }

    @Bean
    public KafkaTemplate kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public ConsumerFactory consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs(){
        return Demo1Config.consumerConfigs(ByteArrayDeserializer.class);
    }
}
